package com.drondea.testclient.service.provider;

import com.drondea.sms.message.ILongSMSMessage;
import com.drondea.sms.message.IMessage;
import com.drondea.sms.message.MessageProvider;
import com.drondea.testclient.util.CommonThreadPoolFactory;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 客户端消息提供者抽象类
 * 
 * @author shd
 * @date 2022/7/15 15:16
 */

@Slf4j
public abstract class AbstractClientMessageProvider implements MessageProvider {

    private static final int MAX_QUEUE_SIZE = 50000;

    /**
     * 开启扫描thread的标识
     */
    private AtomicBoolean STARTED = new AtomicBoolean();
    private boolean isLog = false;
    protected final ConcurrentLinkedQueue<ILongSMSMessage> MESSAGES_QUEUE = new ConcurrentLinkedQueue<>();

    public void init() {
        if (STARTED.get()) {
            return;
        }
        //避免线程问题
        if(STARTED.compareAndSet(false, true)) {
            loopPrintQueueSize();
        }
    }
    
    private void loopPrintQueueSize() {
        CommonThreadPoolFactory.getInstance().getScheduleExecutor().schedule(() -> {
            if (STARTED.get()) {
                if (MESSAGES_QUEUE.size() > MAX_QUEUE_SIZE) {
                    log.info("当前队列堆积数大于最大限制，将在稍后重试：{}", MESSAGES_QUEUE.size());
                    isLog = true;
                } else if (MESSAGES_QUEUE.size() > 0) {
                    log.info("当前队列堆积数：{}", MESSAGES_QUEUE.size());
                    isLog = true;
                } else if (isLog) {
                    isLog = false;
                    log.info("当前队列堆积数：{}", MESSAGES_QUEUE.size());
                }
                loopPrintQueueSize();
            } else {
                log.info("队列监控线程运行结束...");
            }
        }, 1000, TimeUnit.MILLISECONDS);
    }

    public boolean addMessage(ILongSMSMessage message) {
        if (MESSAGES_QUEUE.size() > MAX_QUEUE_SIZE) {
            return false;
        }
        return MESSAGES_QUEUE.offer(message);
    }

    @Override
    public void responseMessageMatchFailed(String s, IMessage iMessage) {
        log.error("responseMessageMatchFailed: {}, {}", s, iMessage);
    }
    
    public void destroy() {
        STARTED.set(false);
    }
    
}
